package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: ws
 * Date: 2023-10-01
 * Time: 21:59
 */

/*
 * 通过这个类，针对硬盘上的消息进行管理
 * 1. 设计了目录结构和文件格式
 * 2. 实现了目录的创建和删除
 * 3. 实现了统计文件的读写
 * 4. 实现了消息的写入
 * 5. 实现了消息的删除->随机访问文件
 * 6. 实现了加载所有的消息
 * 7. 实现了垃圾回收
 *
 * data 的目录结构
 *
 * data // 目录                                               长度固定使用4个字节记录，消息数据跟随在后面
 *      meta.db
 *      testQueue1 // 目录，存储消息队列的文件信息        4byte       20byte     4byte        30byte    4byte        80byte
 *          queue_data.txt // 保存消息的内容     格式：[消息的长度|消息的二进制数据|消息的长度|消息的二进制数据|消息的长度|消息的二进制数据]
 *          queue_stat.txt                    消息的二进制数据 = [basicProperties|body|isValid]
 *      testQueue2
 *          queue_data.txt                             格式：总消息数量\t有效消息数量
 *          queue_stat.txt // 保存消息的统计信息           形式：2000\t1500 （用于触发 GC，GC 的解决方案：当消息的数目超过 2000，并且有效消息数目低于总消息数目的 50%，就触发一次 GC
 */
public class MessageFileManger {
    // 定义一个内部类，来表示该队列的统计信息
    static public class Stat {
        // 直接使用 public，类似于 C 的结构体，方便使用
        public int totalCount;// 总消息数量
        public int validCount;// 有效消息数量
    }

    public void init() {
        // 为了代码风格一致，暂时并没有需要初始化的东西
    }


    // 约定消息文件所在的目录和文件名
    // 获取到指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    /* 获取队列的消息数据文件路径
     * 二进制文件使用 txt 作为后缀，不是太合适，txt 一般表示文本（但是也能使用）
     * .bin, .dat 更合适
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    /*
     * 获取该队列的消息统计文件路径
     */
    public String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName) {
        // 由于当前的消息统计文件是文本文件，可以直接使用 Scanner 来读取
        Stat stat = new Stat();
        try (FileInputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner = new Scanner(inputStream);
            // 2000\t1500，读取硬盘中的数据到内存中
            stat.totalCount = scanner.nextInt();// 第一个是消息总数量
            stat.validCount = scanner.nextInt();// 第二个数有效消息的数量
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat) {
        // 使用 PrintWrite 来写文件
        // OutputStream 打开文件，默认情况下，会直接把源文件清空，此时相当于新的数据覆盖了旧的数据
        // OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))，编译器推荐下面的方法
        try (OutputStream outputStream = Files.newOutputStream(Paths.get(getQueueStatPath(queueName)))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            // 将内存中的新数据，更新到硬盘中
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /*
     * 创建队列对应的文件和目录
     */
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 先创建对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 不存在，就创建
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败!baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败！queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败！queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给消息统计文件设置初始值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    /*
     * 删除队列的目录和文件
     */
    public void destroyQueueFiles(String queueName) throws IOException {
        // 先删除里面的文件，再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            // 任意一个失败，整体就失败
            throw new IOException("[删除队列的目录和文件失败！ baseDir=" + baseDir.getAbsolutePath());
        }
    }

    /*
     * 检查队列的目录和文件是否存在
     * 后续需要持久化，就要写到文件中，就需要检查文件是否存在
     */
    public boolean checkFilesExist(String queueName) {
        // 判定队列的数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    /*
     * 这个方法用来把一个新的消息，放到对应的文件中
     * queue 表示要写入的队列，message 表示要写的信息
     */
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1. 检查一下当前要写入的队列对应的文件是否存在
        if (!checkFilesExist(queue.getName())) {
            throw new MqException("[MessageFileManger] 队列对应的文件不存在！queueName=" + queue.getName());
        }
        // 2. 把 Message对象 进行序列化
        byte[] messageBinary = BinaryTool.toBytes(message);
        /*
         * 保证线程的安全
         * 以队列为对象进行加锁
         * 如果两个线程，是往同一个队列中写消息，此时需要阻塞等待
         * 如果两个线程，往不同的队列中写消息，此时不需要阻塞等待（不同队列对应不同的文件，各写各的，不会冲突）
         */
        synchronized (queue) {// idea 给出警告，是因为它不确定这个形参是否被多个线程共享，如果不是，加锁就没有用了
            // 3. 先获取到当前的队列数据文件的长度，用这个来计算出该 Message对象 的 offsetBegin 和 offsetEnd
            // 把新的 Message数据，写入到队列数据文件的末尾，此时的 Message对象 的offsetBegin，就是 当前文件的长度 + 4
            // offsetEnd 就是 当前文件长度 + 4 + message.length
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // 通过这个方法 queueDataFile.length() 得到文件的长度，单位 byte
            message.setOffsetBegin(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 4. 写入消息到文件,true 表示追加到文件末尾
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                // 根据规定的格式
                // 先写入当前消息的长度，栈 4 个字节
                // outputStream.write(messageBinary.length); 该方法的参数虽然是 int，但是一次只能写入一个 byte，而我们规定的是 4 个byte
                // 解决方法：
                // 1) 我们 int 拆分为 4 个 byte，连续写入四次
                // outputStream.write(num & 0xff);
                // outputStream.write((num >> 8) & 0xff);
                // outputStream.write((num >> 16) & 0xff);
                // outputStream.write((num >> 24) & 0xff);
                // 2) 使用 Java 标准库的现成的类，DataOutputStream/DataInputStream
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 实打实的写四个byte
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount++;
            stat.validCount++;
            writeStat(queue.getName(), stat);
        }
    }

    /*
     * 删除消息的方法
     * 逻辑删除，将硬盘数据里的 isValid 属性，设置为 0x0
     * 1. 先把文件中的这一段数据读出来，反序列化为 Message对象
     * 2. 把 isValid 这个 byte 改成 0x0 （仅仅调整一个 byte 的值，所以序列化后 Message 的长度不变）
     * 3. 把上述数据重新写回文件
     * 所以此时的 Message对象，必须包含有效的 offsetBegin 和 offsetEnd
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        // TODO 不应该检查消目录和息是否存在吗？
        /*
         * FileInputStream 和 FileOutputStream 都是从文件头开始读写
         * 此处需要从指定位置进行读写，随机访问
         * 所以需要使用 RandomAccessFile 的 seek 方法（它能够调整光标的位置）
         */
        // mode "rw" 既可以 read，也可以 write
        synchronized (queue) {// 保证线程安全
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 先从文件中队去对应的 Message 数据
                // 开辟空间
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBegin())];
                // 让光标找到起点
                randomAccessFile.seek(message.getOffsetBegin());
                // 开始读取文件，放到 bufferSrc 中
                randomAccessFile.read(bufferSrc);
                // 2. 把当前读取出来的二进制数据，反序列化为 Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 3. 修改 isValid 为 0x0
                diskMessage.setIsValid((byte) 0x0);
                // 4. 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                // 此时需要注意光标的位置已经移动，所以需要重新定位
                // read，write，seek 都会使光标移动
                // 仅仅调整一个 byte 的值，所以序列化后 Message 的长度不变，直接在原位置覆盖即可
                randomAccessFile.seek(message.getOffsetBegin());
                randomAccessFile.write(bufferDest);
            }
            // 更新统计文件，要将有效消息 - 1
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount--;
            }
            writeStat(queue.getName(), stat);
        }
    }


    /*
     * 读取文件中所有的消息内容，加载到内存中
     * 在程序启动时调用，不涉及多线程
     * LInkedList 便于后续的头删尾删
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = Files.newInputStream(Paths.get(getQueueDataPath(queueName)))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)){
                // 记录光标的位置
                long currentOffset = 0;
                // 一个文件中包含了很多的信息，需要循环读取
                while (true) {
                    // 1. 读取消息的长度，这里的 readInt 可能会读取到文件的末尾（EOF）
                    // 而 readInt 方法，不同于一般的方法，当它到达文件末尾，会抛出 EOFException异常
                    int messageSize = dataInputStream.readInt();
                    // 2. 按照这个长度，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);// 返回读取的总长，单位 byte
                    if (messageSize != actualSize) {
                        // 如果不匹配，说明文件有问题，格式错乱了
                        throw new MqException("[MessageFileManager 文件格式错误！queueName=" + queueName);
                    }
                    // 3. 把读到的数据反序列化
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 4. 判定一下这个对象，是不是无效对象
                    if (message.getIsValid() != 0x1) {
                        // 无效数据，直接跳过，但是也要更新光标位置
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    // 5. 有效数据，加入链表中,加入之前需要填写 offsetBegin 和 offsetEnd
                    // 需要知道光标的位置，DataInputStream 并没有提供方法，所以需要手动计算
                    message.setOffsetBegin(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    // 为下一个消息更新位置
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 这个 catch 并非真正的处理异常，而是处理正常的业务逻辑，文件读到末尾，会被 readInt 抛出异常
                // 这个 catch 语句也不需要做啥特殊的处理
                System.out.println("{MessageFileManger] 恢复 Message 数据完成！");
            }
        }
        return messages;
    }

    /*
     * 检查是否要针对该队列的消息数据文件进行 GC
     */
    public boolean checkGC(String queueName) {
        // 判断是否要 GC，根据总消息数和有效消息数，这俩个值都在消息统计文件中
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.totalCount / (double) stat.validCount < 0.5) {
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /*
     * 真正实行消息数据文件的垃圾回收操作
     * 使用复制算法
     * 创建一个新的文件 queue_data_new.txt
     * 把之前的有效数据写到新的文件中
     * 删除旧的文件，在把新的文件改名为 queue_data.txt
     * 最后更新统计文件 queue_stat.txt
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 进行 gc 的时候，针对文件大洗牌，所以其他的线程不能对文件做出任何的修改
        synchronized (queue) {
            // gc 可能比较耗时，统计一下时间
            long gcBegin = System.currentTimeMillis();
            // 1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 正常情况，这个文件不存在，如果存在，说明上次的 gc 进行了一半，出了问题
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在！queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败！queueDataFile=" + queueDataNewFile.getAbsolutePath());
            }
            // 2. 从旧的文件中读取有效的消息
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 3. 把有效的消息拷贝到新文件
            try (OutputStream outputStream = Files.newOutputStream(queueDataNewFile.toPath())) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        // 先写四个字节的长度
                        dataOutputStream.writeInt(buffer.length);
                        // 再写消息
                        dataOutputStream.write(buffer);
                    }
                }
            }
            // 4. 删除旧的文件，并且重新命名新的文件
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("[MessageFileManger] 删除数据文件失败！queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            // 把 queue_data_new.txt => queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("[MessageFileManger] 文件重命名失败！queueDataNewFile=" + queueDataNewFile.getAbsolutePath() + "，queueDataOldFile" + queueDataOldFile.getAbsolutePath());
            }
            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            // 全部都有效
            stat.totalCount = stat.validCount = messages.size();
            writeStat(queue.getName(), stat);
            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManger] gc 执行完毕！queueName=" + queue.getName() + "，time=" + (gcEnd - gcBegin) + "ms");
        }
    }
}
